package com.cwh.taskcenter.task.consumer;

import com.alibaba.cola.exception.BizException;
import com.alibaba.fastjson2.JSON;
import com.cwh.taskcenter.domain.common.LockConstants;
import com.cwh.taskcenter.domain.task.gateway.UserTaskRecordGateWay;
import com.cwh.taskcenter.lock.DistributedLock;
import com.cwh.taskcenter.task.api.TaskServiceI;
import com.cwh.taskcenter.task.dto.event.UserBehaviorEvent;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;


/**
 * @author cwh
 */
@Slf4j
@Component
@AllArgsConstructor
public class TaskEventConsumer {

    private final UserTaskRecordGateWay userTaskRecordGateWay;

    private final TaskServiceI taskService;

    private final DistributedLock distributedLock;


    @KafkaListener(topics = "${kafka.topics.user-behavior}",
            groupId = "${spring.kafka.consumer.group-id}",
            concurrency = "4")
    public void handleUserBehaviorEvent(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        log.info("处理用户行为事件:{}", record);
        String eventMsg = record.value();
        UserBehaviorEvent eventRequest = JSON.parseObject(eventMsg, UserBehaviorEvent.class);
        String eventKey = eventRequest.getUuid(); // 事件唯一标识
        String lockKey = LockConstants.LOCK_KEY_TASK_EVENT_REPORT + eventKey;
        boolean lockAcquired = false;
        boolean processSuccess = false;

        try {
            // 1. 获取分布式锁
            lockAcquired = distributedLock.tryLock(lockKey, 10, 10, TimeUnit.SECONDS);
            if (!lockAcquired) {
                log.warn("获取事件锁失败，可能并发处理: {}", eventKey);
                return;
            }

            // 2. 在锁内进行幂等性检查
            if (isDuplicateEvent(eventKey)) {
                log.warn("检测到重复事件，跳过处理: {}", eventRequest);
                return;
            }
            taskService.handleEvent(eventRequest);
            processSuccess = true;
        } catch (Exception e) {
            log.error("处理用户行为事件异常", e);
            throw new BizException("处理用户行为事件异常");
        } finally {
            // 4. 手动提交偏移量（仅在成功处理时提交）
            if (processSuccess) {
                acknowledgment.acknowledge();
                log.debug("已提交Kafka偏移量: topic={}, partition={}, offset={}",
                        record.topic(), record.partition(), record.offset());
            }

            // 5. 释放分布式锁
            if (lockAcquired) {
                distributedLock.unlock(lockKey);
            }
        }
    }

    private boolean isDuplicateEvent(String eventKey) {
        return userTaskRecordGateWay.existsByUniqueKey(eventKey);
    }
}
